package edu.csl.study.spark.basic

import java.io.File
import java.net.URI
import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.hadoop.io.compress.GzipCodec
import org.apache.hadoop.io.{IntWritable, LongWritable, Text}
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
import org.apache.spark.rdd.{JdbcRDD, RDD}
import org.apache.spark.{HashPartitioner, SparkConf, SparkContext}

import scala.util.parsing.json.JSON

object Core_RDD_Method_Local {
  /**
   * 获取上下文
   *
   * @return
   */
  def newContext:SparkContext={

    /**注意：线程数不能设置为1，必须大于1 .因为必须有一个线程接受数据，其他线程处理数据
     * local[N] ，N为2,3,4,5等数字 或者 local[*]
     * local[*] 表示自动分配资源，和CPU个数一样，本机为4
     */
    val conf:SparkConf = new SparkConf().setAppName("WordCount").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("warn")
    sc
  }
  val rootDir = System.getProperty("user.dir")+ "\\testFile\\"
  def main(args: Array[String]): Unit = {
    //读取数据测试

     println(System.getProperty("user.dir"))
      // 1.配置信息
      val sc = newContext
         //测试各种读取/存储数据的方法
    RDDOperate_Action(sc)
      //2.数据输入 :
      val rdd:RDD[String] = sc.textFile(rootDir+"words.txt",4)
      println("rdd length = "+rdd.partitions.length)

     val rdd1:RDD[String] = rdd.flatMap(x => x.split(" "))
     val rdd1_map = rdd1.map(x => (x,1))
     ////reduceByKey、sortByKey
     val rdd1_reduce:RDD[(String,Int)] = rdd1_map.reduceByKey(_+_)

     rdd1_reduce.collect().foreach(println)
     println("---words---结果："+rdd1_reduce.collect().mkString(","))

     val rdd2 = sc.parallelize(List(5, 6, 4, 3))
     val rdd3 = sc.makeRDD(List(1, 2, 3, 4))
     //求交集
     val rdd21 = rdd2.mapPartitionsWithIndex((index,iter) => {

      iter.map(x => {
        println("第"+index+"个分区数据： "+x )
        x * 10
      } )
    })
    println("---按分区mapPartitionsWithIndex--partitions.length = "+rdd2.partitions.length)
    //打印结果
    println("---按分区mapPartitionsWithIndex---结果："+rdd21.collect().mkString(","))
    println("---交集intersection---partitions.length = "+rdd2.partitions.length)
    println("---交集intersection-- -结果："+rdd2.intersection(rdd3).collect().mkString(","))
    //求并集
    println("---并集union---partitions.length = "+rdd3.partitions.length)
    println("---并集union---结果："+rdd2.union(rdd3).collect().mkString(","))


    val rdd4 = sc.parallelize(List(("tom", 1), ("tom", 2), ("jerry", 3), ("kitty", 2)))
    val rdd5 = sc.parallelize(List(("jerry", 2), ("tom", 1), ("jim", 2)))

    /***
     * 分组cogroup: 在类型为(K,V)和(K,W)的RDD上调用，返回一个(K,(Iterable<V>,Iterable<W>))类型的RDD
     * 结果：
     * (jim,(CompactBuffer(),CompactBuffer(2)))
     * |(tom,(CompactBuffer(1, 2),CompactBuffer(1)))
     * |(jerry,(CompactBuffer(3),CompactBuffer(2)))
     * |(kitty,(CompactBuffer(2),CompactBuffer()))
     */

    println("---cogroup---结果："+rdd4.cogroup(rdd5).collect().mkString("|"))
      //关闭上下文
      sc.stop()
  }

  /**
   * 读取数据
   */
  def readData(sc:SparkContext):Unit = {
    println("*******************读取Demo测试 - 开始 -*******************")
    /**
     * 1.读取多个文件：可以使用通配符，比如*.txt   C:\\*\*.txt   C:\\read*\*.txt
     * 2.textFile读取普通文本数据、Json数据、CSV数据、TSV数据。
     *   2.1 CSV： comma separated values；即“逗号分隔值”，其文件以纯文本形式存储表格数据（数字和文本）
     *       每条记录占一行,以逗号为分隔符,逗号前后的空格会被忽略
     *       字段中包含有逗号，该字段必须用双引号括起来
     *       字段中包含有换行符，该字段必须用双引号括起来
     *       字段前后包含有空格，该字段必须用双引号括起来
     *       csv文件用记事本和excel都能打开，用记事本打开显示逗号，用excel打开，没有逗号了，逗号都用来分列了,还可有Editplus打开。
     *    2.2 TSV：tab separated values；即“制表符分隔值”，即以Tab（\t）隔开。
     *       rdd中使用 _.split("\t")分割
     *    2.3 两种格式都能使用文本或者Excel格式打开，但是如果使用Excel格式打开，会修改内容（提示修改时不点确认就不会修改），所以不推荐。
     *    2.4 由于textFile是一行行读取的，读取的Json数据，一个完整的Json必须放在一行中。
     *  3.读取SequenceFile格式的数据：sc.sequenceFile()
     *  4.读取Object格式的数据：sc.objectFile()
     *  5.通过Hadoop API读取HDFS上的数据
     */
    println("*******************读取txt*******************")
    val rdd_txt:RDD[String] = sc.textFile(rootDir+"words.txt",4)
    println("*******************读取Json*******************")
    val rdd_json = sc.textFile(rootDir+"person.json",4)
    println(rdd_json.collect().mkString("\n"))
    println("-----json格式字符串转成Map-----")
    val content=  rdd_json.map(JSON.parseFull)
    content.foreach({
        case Some(map) => println(map)
        case None => println("无效的Json")
        case _ => println("其他异常")
    })
    println("*******************读取objectFile*******************")
    //注意，要先执行saveData 把Person存储到saveObject才能执行成功。
    val rdd_object = sc.objectFile[Person](rootDir+"saveObject",1)
    rdd_object.foreach(println)

    println("*******************显式调用Hadoop AIP-newAPIHadoopFile*******************")
    val path = "hdfs://192.168.100.20:9000/practise/kaikeba/txt/user.txt"
    /*  第一个参数LongWritable:每一行的偏移量
        第二个参数 Text： 偏移量对应的内容
        第三个参数TextInputFormat： 读取数据的InputFormat
     */
    val  newAPIHadoopFile:RDD[(LongWritable,Text)] = sc.newAPIHadoopFile[LongWritable,Text,TextInputFormat](path,classOf[TextInputFormat]
    ,classOf[LongWritable],classOf[Text])
    val result = newAPIHadoopFile.map(_._2.toString).collect()
    println(result.mkString("\n"))

    println("*******************读取MySQL数据*******************")
    val querySQL = "select* from person where  id >= ? and id <= ? ";
    val MysqlURL = "jdbc:mysql://127.0.0.1:3306/bigdata?useUnicode=true&characterEncoding=utf8";
    var inputMySQL = new JdbcRDD(sc,
      ()=>{
         DriverManager.getConnection(MysqlURL,"root","admin")
      },
      querySQL,
      4,
      10,
      1,
      r =>(r.getInt("id"),
        r.getString("name"),
        r.getInt("age"),
        r.getInt("sex"),
        r.getString("phone")
      ))
    inputMySQL.foreach(println)

    println("*******************读取Demo测试 - 结束 -*******************")
  }

  def RDDOperate_transform(sc:SparkContext):Unit = {
    println("*******************RDD转换（Transformations）算子 - 开始 -*******************")
    val rdd1:RDD[Int] = sc.parallelize(1 to 10,3)
    /**
     * RDD[Int] -> Array[RDD[Int]]
     * randomSplit： 根据权重（weights）对RDD进行拆分，生成几个RDD取决于Array的长度。
     * 比如Array(1,4,5)
     *   1:4:5的元素个数比例生成3个RDD
     */
    println("*******************RDD转换（Transformations）算子 - randomSplit -*******************")
    val splitRDD:Array[RDD[Int]] = rdd1.randomSplit(Array(1,4,5))
    /**
     * RDD[Int] -> RDD[Array[Int]]
     * glom()： 把RDD中每一个分区的数据变成一个数组类型。
     *
     */
    println("*******************RDD转换（Transformations）算子 - glom -*******************")
    val rddGlom:RDD[Array[Int]] = rdd1.glom()
    /**
     * RDD[Int] -> RDD[Array[Int]]
     * 拉链操作- zip(other:RDD[T])： 将两个RDD中的元素，以键值对的形式进行合并，Key为第一个RDD的元素，Value为第二个RDD中的元素。
     * 要求：两个RDD的元素个数和分区个数一样，否则会出现异常。
     * 比如：
     * RDD1
     *    分区1：1 2
     *    分区2：3 4
     * RDD2
     *    分区1：5 6
     *    分区2：7 8
     * 结果：
     *    分区1：(1,4) (2,6)
     *    分区2：(3,7) (4,8)
     *
     *  另外：zipPartitions()
     */
    println("*******************RDD转换（Transformations）算子 - zip -*******************")
    val rddZip1:RDD[Int] = sc.parallelize(1 to 3,3)
    val rddZip2:RDD[String] = sc.parallelize(Array("A","B","C"),3)
    var result1  = rddZip1.zip(rddZip2)
    println(result1.collect().mkString(","))
    /**
     * zipPartitions()
     * 优势：可以利用函数自定义zip合并逻辑
     */
    println("*******************RDD转换（Transformations）算子 - zipPartitions -*******************")
    val result2 = rddZip1.zipPartitions(rddZip2)((iter1,iter2) =>{
      var result = List[(Int,String)]()
      while(iter1.hasNext && iter2.hasNext){
        result ::= (iter1.next(),iter2.next())
      }
      result.iterator
    })
    println(result2.collect().mkString(","))
    /**
     * zipWithIndex()：将RDD中的元素和索引进行合并。
     *
     */
    println("*******************RDD转换（Transformations）算子 - zipWithIndex -*******************")
     var result3:RDD[(String,Long)]  = rddZip2.zipWithIndex()
    println(result3.collect().mkString(","))

    /**
     * zipWithUniqueId()：将RDD中的元素和唯一ID进行zip操作。
     * 元素唯一ID:
     *   X*N +K
     *   n为总分区数，K为当前元素所在分区号，X表示从0开始的自然数（不同分区都是从0开始的）
     *
     */
    println("*******************RDD转换（Transformations）算子 - zipWithUniqueId -*******************")
    var result4:RDD[(String,Long)]  = rddZip2.zipWithUniqueId()
    println(result4.collect().mkString(","))
    println("*******************RDD转换（Transformations）算子 - sortBy -*******************")
    val rdd_sortBy = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),1)
    var result_sotrBy = rdd_sortBy.sortBy(_._2)
    println(result_sotrBy.collect().mkString(","))
    println("*******************RDD转换（Transformations）算子 - sortByKey -*******************")
    result_sotrBy = rdd_sortBy.sortByKey()
    println(result_sotrBy.collect().mkString(","))
    /**
     * partitionBy 设置自定义分区器，这里的例子是官方自带的HashPartitioner。
     */
    println("*******************RDD转换（Transformations）算子 - partitionBy  -*******************")
    val rdd_partitionBy = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),2)
    val result_partitionBy = rdd_partitionBy.partitionBy(new HashPartitioner(6))
    println(result_partitionBy.collect().mkString(","))

    println("*******************RDD转换（Transformations）算子 - reduceByKey  -*******************")
    val rdd_reduceByKey = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),1)
    val result_reduceByKey:RDD[(String,Int)] = rdd_reduceByKey.reduceByKey(_+_)
    println(result_reduceByKey.collect().mkString(","))

    /**
     * foldByKey：和reduce相比，可以传入一个初始值。
     */
    println("*******************RDD转换（Transformations）算子 - foldByKey  -*******************")
    val rdd_foldByKey = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),1)
    val result_foldByKey:RDD[(String,Int)] = rdd_foldByKey.foldByKey(20)(_+_)
    println(result_foldByKey.collect().mkString(","))

    println("*******************RDD转换（Transformations）算子 - groupByKey  -*******************")
    val rdd_groupByKey = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),1)
    val result_groupByKey:RDD[(String, Iterable[Int])] = rdd_groupByKey.groupByKey()
    println(result_groupByKey.collect().mkString(","))

    /**
     * combineByKey解释：
     * def combineByKey[C](
     *   createCombiner: V => C,     //第一次访问时会调用，生成初始的C类型值。（C和V也可以同一类型）
     *   mergeValue: (C, V) => C,    //把C类型和V类型聚合成C类型。
     *   mergeCombiners: (C, C) => C //每个分区有独立的Combiner，这里是定义多个分区之间的聚合方式
     * ): RDD[(K, C)]
     *
     * 注意：aggregateByKey、foldByKey以及reduceByKey底层都使用了combineByKey
     *      foldByKey和reduceByKey输入和输出的类型必须一致，而aggregateByKey和combineByKey不必
     */
    println("*******************RDD转换（Transformations）算子 - combineByKey  -*******************")
    val rdd_combineByKey = sc.parallelize(List(("张三",90f),("张三",92f),("张三",91f),("李四",88f),("李四",99f),("王五",91f),("王五",94f)),1)
    val result_combineByKey
    = rdd_combineByKey.combineByKey(
      grade =>(grade,1),  //
      (gc:(Float,Int),grade) => (gc._1 + grade,gc._2 +1),
      (gc1:(Float,Int),gc2:(Float,Int)) =>(gc1._1 + gc2._1,gc1._2 + gc2._2)
    )
    //结果为(张三,(273.0,3)),(李四,(187.0,2)),(王五,(185.0,2))
    println("combineByKey结果："+result_combineByKey.collect().mkString(","))
    //利用上面的结果求平均值
    val result_combineByKey_map = result_combineByKey.map(t => (t._1,t._2._1/t._2._2))
    println("平均值:"+result_combineByKey_map.collect().mkString(","))

    /**
     * aggregateByKey底层是调用了combineByKey，简化了combineByKey的操作
     * def aggregateByKey[U: ClassTag]
     * (zeroValue: U)      //初始值，底层利用combineByKey的createCombiner实现
     * (
     *  seqOp: (U, V) => U, //把V聚合到U对象中
     *  combOp: (U, U) => U //每个分区有独立的Combiner，这里是定义多个分区之间的聚合方式
     *
     * ): RDD[(K, U)]
     */
    println("*******************RDD转换（Transformations）算子 - aggregateByKey  -*******************")
    val rdd_aggregateByKey: RDD[(String, Float)] = sc.parallelize(List(("张三",90f),("张三",92f),("张三",91f),("李四",88f),("李四",99f),("王五",91f),("王五",94f)),1)
    val result_aggregateByKey:RDD[(String,scala.collection.mutable.Set[Float])] =
      //初始值是一个Set[Float]类型
      rdd_aggregateByKey.aggregateByKey(scala.collection.mutable.Set[Float]())(
      (scoreSet:scala.collection.mutable.Set[Float],score:Float) => {
          scoreSet += score  //定义Set[Float]与Float值是怎么聚合的。
      },
      (scoreSet1:scala.collection.mutable.Set[Float],scoreSet2:scala.collection.mutable.Set[Float]) => {
          scoreSet1 ++= scoreSet2//定义多个Set[Float]时怎么聚合的
      }
     )
    println(result_aggregateByKey.collect().mkString(","))
    //结果:(张三,Set(92.0, 91.0, 90.0)),(李四,Set(88.0, 99.0)),(王五,Set(91.0, 94.0))
    /**
     * mapValues方法：对每个value进行map操作。
     *   这里每个value + 100
     */
    println("*******************RDD转换（Transformations）算子 - mapValues  -*******************")
    val rdd_mapValues = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),1)
    val result_mapValues:RDD[(String, Int)] = rdd_mapValues.mapValues(_+100)
    println(result_mapValues.collect().mkString(","))

    /**
     * flatMapValues方法：对value值进行flatMap操作
     */
    println("*******************RDD转换（Transformations）算子 - flatMapValues  -*******************")
    val rdd_flatMapValues = sc.parallelize(List(("张三","1 1 1"),("李四","2 3 4"),("王五","2 999 3"),("张三","555 3 7")),1)
    val result_flatMapValues:RDD[(String, String)] = rdd_flatMapValues.flatMapValues(_.split(" "))
    println(result_flatMapValues.collect().mkString(","))
    //结果：(张三,1),(张三,1),(张三,1),(李四,2),(李四,3),(李四,4),(王五,2),(王五,999),(王五,3),(张三,555),(张三,3),(张三,7)


    println("*******************RDD转换（Transformations）算子 - 结束 -*******************")
  }

  /**
   * Action算子
   *  collect：将所有元素转换成数组
   *  first：第一个元素
   *  take：返回N个元素组成的数组。
   *  top : 默认先将元素按降序排列（可以通过参数修改），再返回前N个元素。
   *  takeOrdered 返回N个元素，第二个参数指定排序规则。top方法调用的就是takeOrdered第二个参数为降序。
   *  reduce
   *  aggregate
   *  fold
   *  foreach
   *  foreachPartition
   *  count：元素个数
   *  sortBy
   *
   * 键值对行动操作
   *   lookup(key):Seq[V] : 指定RDD中K的值，返回K对应的Value的集合封装在Seq序列中。
   *   countByKey()：Map[K,Long]：每一个Key的次数组成的Map
   *
   *
   * @param sc
   */
  def RDDOperate_Action(sc:SparkContext):Unit = {
    println("*******************RDD行动（Action）算子 - 开始 -*******************")
    val rdd_foldByKey = sc.parallelize(List(("张三",19),("李四",18),("王五",26),("张三",40)),1)
    val result_foldByKey:RDD[(String,Int)] = rdd_foldByKey.foldByKey(20)(_+_)
    println(result_foldByKey.collect().mkString(","))


    println("*******************RDD行动（Action）算子 - 结束 -*******************")


  }
  /**
   *
   * 存储数据
   */
  def savaData(sc:SparkContext):Unit = {
    println("*******************存储Demo测试 - 开始 -*******************")
    /***
     * 1.Txt CSV  TSV文件都是使用saveAsTextFile
     *   CSV 数据分隔符 逗号(,) 把数据格式转换后的数据，元素以逗号隔开
     *   TSV 数据分隔符 TAB(\t) 把数据格式转换后的数据，元素以TAB隔开
     * 2. rdd.saveAsSequenceFile(savePath,Some(classOf[GzipCodec]))
     *   GzipCodec表示压缩格式
     * 3.saveAsObjectFile
     * 4.saveAsHadoopFile/saveAsNewAPIHadoopFile
     *   saveAsHadoopFile 是旧版本的，现在一般使用新版本apisaveAsNewAPIHadoopFile
     * 5.存储到Mysql
     *   注意：MySQL无法支撑整个OLAP系统（响应慢），一般只用来处理少量数据。
     *         如果要进行并发和数据量比较高，
     *         那么可以考虑使用NewSQL(TiDB)或者基于Hbase的Phoenix框架
     *   批量存储，见：Core_MYSQL_Scala_local.scala
     */
    println("*******************存储Demo测试 - TXT -*******************")
    val rdd_Txt = sc.parallelize(List(("张三",19),("李四",18),("王五",26)),1)
    var savePath = rootDir+"saveTxt"
    var file =  new File(savePath)
    FileTools.deleteDir(file)

    rdd_Txt.saveAsTextFile(savePath)

    val array =  Array(19,18,17,26)
    println("*******************存储Demo测试 - CVS -*******************")
    var rdd_CSV = sc.parallelize(Array(array.mkString(",")),1)
      savePath = rootDir+"saveCSV";
      file =  new File(savePath)
      FileTools.deleteDir(file)
      rdd_CSV.saveAsTextFile(savePath)
    println("*******************存储Demo测试 - TSV -*******************")
    var rdd_TSV = sc.parallelize(Array(array.mkString("\t")),1)
    savePath = rootDir+"saveTSV";
    file =  new File(savePath)
    FileTools.deleteDir(file)
    rdd_TSV.saveAsTextFile(savePath)

    println("*******************存储Demo测试 - SequenceFile -*******************")
    val rdd_sequence = sc.parallelize(List(("张三",19),("李四",18),("王五",26)),1)
    savePath = rootDir+"saveSequence"
    file =  new File(savePath)
    FileTools.deleteDir(file)
    rdd_sequence.saveAsSequenceFile(savePath,Some(classOf[GzipCodec]))

    println("*******************存储Demo测试 - Object -*******************")

    val rdd_Object = sc.parallelize(List(Person("张三",19),Person("李四",18),Person("王五",26)),1)
    savePath = rootDir+"saveObject"
    file =  new File(savePath)
    FileTools.deleteDir(file)
    rdd_Object.saveAsObjectFile(savePath)

    println("*******************存储Demo测试 - saveAsNewAPIHadoopFile -*******************")
    val rdd_Hadoop = sc.parallelize(List(("张三",19),("李四",18),("王五",26)),1)
    savePath =   "hdfs://192.168.100.20:9000/output/saveAsHadoopFile"

    FileTools.deleteHDFSDir(savePath)//先删除路径，方便重复使用

    rdd_Hadoop.saveAsNewAPIHadoopFile(savePath,classOf[Text],classOf[IntWritable], classOf[TextOutputFormat[Text,IntWritable]])
    println("*******************存储Demo测试 - save IN MYSQL -*******************")
    //`name`, `pass`, `age`, `sex`, `phone`
    val rdd_Mysql = sc.parallelize(List(("张三","123456",19,1,"13631666544"),("李四","123456",45,1,"13631666555"),("王五","123456",22,0,"13631666533")),3)
    val MysqlURL = "jdbc:mysql://127.0.0.1:3306/bigdata?useUnicode=true&characterEncoding=utf8";
    rdd_Mysql.foreachPartition(iter =>{
      val connection : Connection = DriverManager.getConnection(MysqlURL,"root","admin")
      val SQL = "INSERT INTO `bigdata`.`person` (`name`, `pass`, `age`, `sex`, `phone`, `remark`) VALUES (?, ?, ?, ?, ?,?)"
      try {
        val pstm: PreparedStatement = connection.prepareStatement(SQL)
        iter.foreach(item => {
          pstm.setString(1, item._1)
          pstm.setString(2, item._2)
          pstm.setInt(3, item._3)
          pstm.setInt(4, item._4)
          pstm.setString(5, item._5)
          pstm.setString(6, "Core_RDD_Method_Local.scala save IN MYSQL  测试")
          pstm.executeUpdate()
        })
      }catch  {
        case e:Exception => e.printStackTrace()
      }finally {
        if(connection !=null) {
          connection.close()
        }
      }

    })


    println("*******************存储Demo测试 - 结束 -*******************")

  }
  case class Person(name :String,age:Int)
}

